package com.design;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class AsyncImageProcessor {
    // 线程池配置
    private static final int CPU_CORES = Runtime.getRuntime().availableProcessors();
    private static final int CORE_POOL_SIZE = CPU_CORES * 2;
    private static final int MAX_QUEUE_CAPACITY = 1000;

    // 全局线程池
    private final ThreadPoolExecutor executor;

    // 性能监控
    private final AtomicInteger activeTasks = new AtomicInteger(0);

    public AsyncImageProcessor() {
        this.executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                CORE_POOL_SIZE * 2,  // 允许临时扩展线程
                60L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(MAX_QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    public CompletableFuture<String> processImage(String imageId) {
        activeTasks.incrementAndGet();

        return CompletableFuture
                .supplyAsync(() -> uploadImage(imageId), executor)
                .thenApplyAsync(this::ocrProcessing, executor)
                .thenApplyAsync(this::modelAnalysis, executor)
                .whenComplete((result, ex) -> {
                    activeTasks.decrementAndGet();
                    if (ex != null) {
                        System.err.println("处理失败: " + ex.getMessage());
                    }
                });
    }

    // 模拟耗时操作
    private String uploadImage(String imageId) {
        sleep(2);
        return imageId + "_UPLOADED";
    }

    private String ocrProcessing(String image) {
        sleep(5);
        return image.replace("UPLOADED", "OCR_PROCESSED");
    }

    private String modelAnalysis(String text) {
        sleep(8);
        return text.replace("OCR_PROCESSED", "MODEL_ANALYZED");
    }

    private void sleep(int milliseconds) {
        try {
            TimeUnit.MILLISECONDS.sleep(milliseconds);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // 监控方法
    public void monitor() {
        System.out.println("\n====== 系统监控 ======");
        System.out.println("活跃任务数: " + activeTasks.get());
        System.out.println("线程池状态: "
                + "[活跃线程数: " + executor.getActiveCount()
                + "/核心线程数: " + executor.getCorePoolSize()
                + "/最大线程数: " + executor.getMaximumPoolSize() + "]");
        System.out.println("队列大小: " + executor.getQueue().size());
    }

    public static void main(String[] args) throws InterruptedException {
        AsyncImageProcessor processor = new AsyncImageProcessor();
        List<CompletableFuture<String>> futures = new ArrayList<>();

        // 模拟200并发请求
        for (int i = 0; i < 200; i++) {
            String imageId = "IMG_" + (i + 1);
            futures.add(processor.processImage(imageId));
            if ((i + 1) % 50 == 0) {
                processor.monitor();
                Thread.sleep(1000);  // 模拟请求间隔
            }
        }

        // 等待所有任务完成
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        System.out.println("\n所有任务处理完成！");
        processor.executor.shutdown();
    }
}
